package com.cmbyc.source

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}

import java.sql.{Connection, DriverManager, PreparedStatement}

/**
 *
 * @program: com.cmbyc.source
 * @author: YCLW058
 * @create: 2021-05-18 17:04
 * @decsription:
 *
 * */

//样例类
case class Student(id: Int, name: String, age: Int)

object mysql_2_mysql {
  def main(args: Array[String]): Unit = {
    //1 env
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2 source
    val ds = env.addSource(new MySQLSource()).setParallelism(1)
    //3 transformation
    //4 sink
    ds.addSink(new MySQLSink)
    ds.print()
    //5 execute
    env.execute()
  }

  class MySQLSource() extends RichParallelSourceFunction[Student] {
    //定义连接 编译语句
    var conn: Connection = _
    var ps: PreparedStatement = _

    override def open(parameters: Configuration): Unit = {
      conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/cd_mysql", "root", "root")
      val sql = "select id,name,age from student3"
      ps = conn.prepareStatement(sql)
    }

    var flag = true

    override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {
      val rs = ps.executeQuery()
      while (rs.next()) {
        val id = rs.getInt("id")
        val name = rs.getString("name")
        val age = rs.getInt("age")
        ctx.collect(Student(id = id, name = name, age = age))
      }


    }

    override def cancel(): Unit = ???

    override def close(): Unit = {
      if (conn != null) conn.close()
      if (ps != null) ps.close()
    }


  }


  class MySQLSink() extends RichSinkFunction[Student] {
    // 定义连接 预编译语句
    var conn: Connection = _
    var ps: PreparedStatement = _

    override def open(parameters: Configuration): Unit = {
      conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/cd_mysql", "root", "root")
      val sql = "INSERT INTO `student1`(`name`,`age`)VALUES (?,?)"
      ps = conn.prepareStatement(sql)
    }

    override def invoke(value: Student, context: SinkFunction.Context): Unit = {
      //获取 具体值
     // ps.setInt(1, value.id)
      ps.setString(1, value.name)
      ps.setInt(2, value.age)

      //执行
      ps.executeUpdate();

    }

    override def close(): Unit = {
      if (conn != null) conn.close()
      if (ps != null) ps.close()
    }
  }

}
